// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jhkim

#include "coin2/exchange/sim/order_v1/gateway.h"

#include <memory>
#include <optional>
#include <string>

#include "orio/record/simple.h"
#include "coin2/exchange/binance_base/api_util/rate_tracker.h"
#include "coin2/exchange/bybit/api_util/rate_tracker.h"
#include "coin2/exchange/okex_common/api_util/rate_tracker.h"

namespace coin2::exchange::sim::order_v1 {

BacktestingGateway::BacktestingGateway(
    const std::string& name,
    const MarketExchangeApi& mea,
    const OrderSystemConfig& oss_config,
    const SimConfig2& config,
    const PrivateSubscriberConfig& subscriber_config,
    std::vector<const IProduct*>* products,
    SymbologyCache* symcache,
    IOrderSubscriber* callback)
    : FeatureFeedConverter(&symbol_manager_, &product_cache_),
      mea_(mea),
      subscriber_config_(subscriber_config) {
  if (mea.exchange == ExchangeType::Binance) {
    rate_tracker_.reset(new coin2::exchange::binance::api_util::BinanceRateTracker(
        mea, oss_config.connection_config()));
  } else if (mea.exchange == ExchangeType::Okex) {
    rate_tracker_.reset(new coin2::exchange::okex_common::api_util::OkexRateTracker(
        mea, oss_config.connection_config()));
  }
  base::order::BaseOrderGateway::symcache_ = symcache;
  set_qty_multiplier(config.sim_qty_multiplier());
  ProductRequest req;
  req.set_mea(mea.String());
  for (const auto& acct_config : config.accounts()) {
    req.set_norm(acct_config.relative_norm());
    for (auto&& product :
         ResolveProductRequestFromPotentiallyWildcard(req, acct_config.start_timestamp(), false)) {
      symcache->AddProduct(
          std::unique_ptr<IProduct>(product->Clone()),
          acct_config.start_timestamp());
      products_.push_back(std::move(product));
      auto prod = products_.back().get();
      const Symbol* symbol = GetSymbol(mea, *prod);
      stored_products_[symbol] = prod;

      matchers_.push_back(std::unique_ptr<BacktestingMatcher>(
          new BacktestingMatcher(matchers_.size(), mea, acct_config, prod)));
      auto* acct_matcher = matchers_.back().get();
      sim_account_ids_.insert(acct_config.sim_account_id());
      matchers_map_[std::make_pair(acct_config.sim_account_id(), symbol)] = acct_matcher;
      feeders_[symbol].push_back(acct_matcher);

      if (osm_.count(acct_config.sim_account_id()) == 0) {
        osm_[acct_config.sim_account_id()].reset(new OrderStateManager(name, mea));
        auto* osm = osm_.at(acct_config.sim_account_id()).get();
        osm->oum.SetOrderManager(&osm->ocm);
      }
      auto* osm = osm_.at(acct_config.sim_account_id()).get();
      if (!osm->oum.is_symbol_registered(prod->native_symbol())) {
        osm->oum.AddProduct(prod);
        osm->oum.AddNativeSymbol(prod->native_symbol(), callback);
        osm->oum.GetProductOrderInfoMutable(prod->native_symbol())
            ->SetInitFillPosition(acct_config.initial_position(), GetCurrentTimestamp());
      }
      if (acct_config.has_max_position_quote()) {
        if (osm->max_position_quote.has_value()) {
          CHECK_EQ(*osm->max_position_quote, acct_config.max_position_quote());
        }
        osm->max_position_quote = acct_config.max_position_quote();
      }
      UpdateAccountInfo(0, symbol);
    }
  }
  switch (mea.market) {
    case MarketType::Spot:
      account_info_update_interval_ = subscriber_config.query_account_balance_period_sec() * 1e9;
      break;
    case MarketType::Futures:
      account_info_update_interval_ = subscriber_config.query_account_position_period_sec() * 1e9;
      break;
    default:
      break;
  }
  InitWithMea(mea_);
  need_exchange_id_to_cancel_ = false;
}

void BacktestingGateway::CheckAndUpdateOrderSpec(
    const coin2::exchange::base::order::OrderSpec& spec) {
  if (spec.duration == coin::proto::OrderDuration::IOC_ORDER && !is_supporting_ioc_) {
    spec.ResetDuration();
  }
  if (!is_supporting_post_only_) {
    spec.ResetPostOnly();
  }
}

bool BacktestingGateway::CheckRateLimit(ActionType action_type) {
  int64_t timestamp = 0L;
  if (last_time_.has_value()) {
    timestamp = *last_time_;
  } else if (last_feed_ts_.has_value()) {
    timestamp = *last_feed_ts_;
  } else {
    return true;
  }
  if (rate_tracker_ && timestamp) {
    SPDLOG_INFO_EVERY_NS(
        60'000'000'000L,
        timestamp,
        "{} {} RATE LIMIT INFO: {}",
        timestamp,
        RequestType_Name(action_type),
        rate_tracker_->GetMaxRateUsedRatio(timestamp, action_type));
    if (rate_tracker_->HasEnoughRateLimit(timestamp, action_type)) {
      rate_tracker_->Push(1/*num_reqs*/, timestamp, action_type);
    } else {
      // TODO(jhkim): use ORDER_REJECTED later
      LOG(ERROR) << RequestType_Name(action_type) + " RATE LIMIT EXCEEDED";
      return false;
    }
  }
  return true;
}

std::optional<int64_t> BacktestingGateway::SubmitOrderForSim(
    int64_t account_id,
    const coin2::exchange::base::order::OrderSpec& spec) {
  if (enable_api_rate_limit_check_ && !CheckRateLimit(ActionType::PLACE)) return std::nullopt;
  std::optional<double> submit_max_rate_used_ratio, cancel_max_rate_used_ratio;
  if (rate_tracker_) {
    if (last_time_.has_value()) {
      submit_max_rate_used_ratio = rate_tracker_->GetMaxRateUsedRatio(*last_time_, ActionType::PLACE);
      cancel_max_rate_used_ratio = rate_tracker_->GetMaxRateUsedRatio(*last_time_, ActionType::CANCEL);
    } else if (last_feed_ts_.has_value()) {
      submit_max_rate_used_ratio = rate_tracker_->GetMaxRateUsedRatio(*last_feed_ts_, ActionType::PLACE);
      cancel_max_rate_used_ratio = rate_tracker_->GetMaxRateUsedRatio(*last_feed_ts_, ActionType::CANCEL);
    }
  }
  // assume that mea equals..
  ProductRequest req;
  auto& osm = *osm_.at(account_id);
  if (spec.direction() == OrderDirection::DIRECTION_BUY && osm.max_position_quote.has_value()) {
    double max_cash = osm.max_position_quote.value();
    double required_cash = spec.price * spec.qty;
    double used_cash = osm.pos_quote + osm.working_quote;
    if (max_cash < used_cash + required_cash) {
      // TODO(jhkim): use ORDER_ERROR later
      LOG(ERROR) << "LACK OF CASH: " << max_cash << " < " << osm.pos_quote << " + "
                 << osm.working_quote << " + " << required_cash;
      return std::nullopt;
    }
  }
  if (!base::order::BaseOrderGateway::IsAcceptableValue(spec)) {
    return std::nullopt;
  }
  req.set_mea(osm.oum.mea().String());
  req.set_native(spec.product);
  auto prod = ResolveProductRequest(req, spec.order_created_time);
  const Symbol* symbol = GetSymbol(osm.oum.mea(), *prod);
  auto* matcher = matchers_map_.at(std::make_pair(account_id, symbol));
  CheckAndUpdateOrderSpec(spec);
  if (last_time_.has_value()) {
    matcher->UpdateTimeOnly(*last_time_);
  }
  matcher->SendOrder(++ioid_, last_time_, spec, submit_max_rate_used_ratio, cancel_max_rate_used_ratio);
  CHECK(!osm.ocm.HasProcOrderId(ioid_)) << ioid_;
  osm.ocm.AddOrderContext(std::make_unique<OrderContext>(ioid_, spec));
  osm.ocm.FindByProcOrderId(ioid_)->HandleOrderSubmitted(spec.order_created_time);
  ioid_map_[ioid_] = std::make_pair(matcher, ioid_);
  return ioid_;
}

bool BacktestingGateway::CancelOrder(int64_t proc_order_id) {
  if (enable_api_rate_limit_check_ && !CheckRateLimit(ActionType::CANCEL)) return false;
  auto ioid_pair = ioid_map_.at(proc_order_id);
  auto* matcher = ioid_pair.first;
  matcher->CancelOrder(ioid_pair.second);
  return true;
}

bool BacktestingGateway::CancelOrderAll(const std::string& symbol) {
  CHECK(false) << "TODO(daniel): CancelOrderAll";
  return true;
}

auto BacktestingGateway::AmendOrder(int64_t proc_order_id, double amend_price, double amend_qty)
    -> std::optional<int64_t> {
  // TODO(daniel):
  CHECK(false) << "TODO(daniel): AmendOrder";
  // auto ioid_pair = ioid_map_.at(proc_order_id);
  // auto* matcher = ioid_pair.first;
  // return matcher->AmendOrder(ioid_pair.second, spec);
  return {};
}

void BacktestingGateway::UpdateAccountInfo(int64_t ts, const Symbol* symbol) {
  for (auto* matcher : feeders_.at(symbol)) {
    auto& osm = *osm_.at(matcher->account_id());
    double net_pos = matcher->GetNetPosition();
    auto* acct_info = osm.oum.GetAccountInfoMutable();
    const auto& product_str = symbol2product_str_.at(symbol);
    osm.oum.GetOrderUpdateMutable()->sim_account_id = matcher->account_id();
    switch (osm.oum.mea().market) {
      case MarketType::Futures: {
        auto pos = matcher->GetMutablePosition();
        pos->set_net_position(net_pos);
        if (net_pos > 0) {
          pos->set_available_long_position(net_pos);
          pos->set_long_position(net_pos);
        } else if (net_pos < 0) {
          pos->set_available_short_position(-net_pos);
          pos->set_short_position(-net_pos);
        }
        acct_info->SetPosition(pos->symbol(), *pos);
        break;
      }
      case MarketType::Spot: {
        auto ccy_bal = matcher->GetMutableBalance();
        ccy_bal->set_total(net_pos);
        ccy_bal->set_available(net_pos);
        acct_info->SetBalance(ccy_bal->currency(), *ccy_bal);
        break;
      }
      default:
        NOTREACHED() << osm.oum.mea().market;
    }
    osm.ocm.ClearFillNotSyncedPos(product_str);
    if (ts - last_account_info_update_ts_[matcher->account_id()] > account_info_update_interval_) {
      osm.oum.GetOrderUpdateMutable()->SetTimestamp(ts);
      osm.oum.PublishAccountInfo();
      last_account_info_update_ts_[matcher->account_id()] = ts;
    }
    osm.UpdatePosition(matcher, symbol);
  }
}

void BacktestingGateway::onTradeFeed(const FeedUpdate& update) {
  last_feed_ts_ = update.timestamp();
  const Symbol* symbol = GetSymbol(update);
  if (!symbol) return;
  if (feeders_.count(symbol) > 0) {
    UpdateAccountInfo(update.timestamp(), symbol);
  }
  onFeed(symbol, update);
  if (!feed_converted) return;
  if (feeders_.count(symbol) == 0) return;
  for (auto* feeder : feeders_.at(symbol)) {
    feeder->UpdateFeed(symbol, last_feed_msg);
  }
}

void BacktestingGateway::onBookFeed(const FeedUpdate& update) {
  last_feed_ts_ = update.timestamp();
  const Symbol* symbol = GetSymbol(update);
  if (!symbol) return;
  if (feeders_.count(symbol) > 0) {
    UpdateAccountInfo(update.timestamp(), symbol);
  }
  onFeed(symbol, update);
  if (exchange_timestamp_override_.has_value()) {
    last_feed_msg.set_exchange_time(exchange_timestamp_override_.value());
    exchange_timestamp_override_.reset();
  }
  if (!symbol || !feed_converted) return;
  if (feeders_.count(symbol) == 0) return;
  for (auto* feeder : feeders_.at(symbol)) {
    if (has_heartbeat_) {
      feeder->UpdateTimeOnly(update.timestamp());
    }
    auto& osm = *osm_.at(feeder->account_id());
    feeder->UpdateFeed(symbol, last_feed_msg);
    auto order_records = feeder->GetOrderRecords();
    for (const auto& record : order_records) {
      LOG(INFO) << update.timestamp() << " " << record.ShortDebugString();
      osm.UpdateEachRecord(update, record, feeder->account_id());
    }
    feeder->WriteAndFlushRecords(first_write_);
    first_write_ = false;
  }
  if (has_heartbeat_) {
    stored_books_[symbol].reset(new FeedUpdate(update.system_name(), update.mea()));
    stored_books_[symbol]->SetBook(update.GetBookBuilder());
    stored_books_[symbol]->SetProduct(*stored_products_[symbol]);
  }
}

void BacktestingGateway::onHeartbeatFeed(const FeedTimer& timer) {
  has_heartbeat_ = true;
  for (auto& [symbol, update] : stored_books_) {
    update->SetTimestamp(timer.Timestamp());
    exchange_timestamp_override_ = timer.Timestamp();
    onBookFeed(*update);
  }
}

const Symbol* BacktestingGateway::GetSymbol(
    const MarketExchangeApi& mea,
    const base::order::IProduct& product) {
  std::string ex = coin::proto::ExchangeType_Name(mea.exchange);
  if (mea.market == MarketType::Futures) ex += "Futures";
  const auto& native_symbol = product.native_symbol();
  const auto& absolute_norm = product.absolute_norm();
  const Symbol* symbol = symbol_manager_.FromString(ex, native_symbol);
  if (!symbol) {
    symbol_manager_.AddSymbol(ex, native_symbol, ex + "_" + native_symbol);
    RegisterSymbols();
    symbol = symbol_manager_.FromString(ex, native_symbol);
    CHECK(symbol);
    symbol2product_str_[symbol] = absolute_norm;
    base_currency_map_[absolute_norm] = product.base().currency();
    LOG(INFO) << ex << " " << absolute_norm << " registered";
  }
  return symbol;
}

}  // namespace coin2::exchange::sim::order_v1
